<?php

namespace app\base\command;


use app\base\model\RedianData;
use app\base\model\RedianTask;
use app\base\service\MemberService;
use app\base\service\TaskService;
use think\admin\Command;
use think\console\Input;
use think\console\Output;
/**
 * 关键词采集任务运行v2版(从指定节点采集数据)
 *
 * 分时段运行任务 采集数量根据产出数量计算
 * Class SmsNum
 * @package app\common\command
 */
class TaskRunV2 extends Command
{
    /**
     * 配置指令
     */
    protected function configure()
    {
        $this->setName('xbase:TaskRunV2');
        $this->setDescription('TaskRunV2');
    }

    /**
     * 执行任务队列处理主函数
     *
     * 该方法负责处理任务队列的完整生命周期，包含四个核心阶段：
     * 1. 数据采集处理（progress 0-2）
     * 2. AI文本生成（progress 3）
     * 3. 音频生成（progress 4）
     * 4. 视频生成（progress 5）
     *
     * 根据任务进度字段(progress)跳转不同处理阶段，实时更新任务状态到数据库，
     * 每个阶段包含资源检查、服务调用、结果验证等完整流程
     *
     * @param Input $input 命令行输入参数对象
     * @param Output $output 命令行输出处理对象
     *
     * 关键流程说明：
     * - 使用$progress字段控制阶段跳转
     * - 每个阶段开始前进行蓝豆余额校验(checkBeans)
     * - 各阶段失败时更新error_msg和error_status字段
     * - 支持蝉镜/HeyGen等第三方平台的特殊处理逻辑
     */
    protected function execute(Input $input, Output $output)
    {
        // 获取队列中的数据
        $queue_data = $this->queue->data;
        // 提取任务ID
        $task_id = $queue_data['task_id'];
        // 初始化任务运行日志ID
        $task_run_log_id = 0;
        // 如果队列数据中包含任务运行日志ID，则赋值给变量
        if(isset($queue_data['task_run_log_id'])){
            $task_run_log_id = $queue_data['task_run_log_id'];
        }
        // 设置队列任务开始执行的状态
        $this->setQueueProgress("开始执行任务", 0);
        // 根据任务ID查询任务信息
        $task = RedianTask::mQuery()->where([ "id"=>$task_id])->find();
        // 如果任务不存在，则设置错误状态
        if(empty($task)){
           // $this->error('任务不存在');
            $this->setQueueError("任务不存在");
        }
        // 获取任务的进度
        $progress = $task['progress'];
        // 根据任务进度执行相应的步骤
        if($progress==0 || $progress==1 || $progress==2){
            $this->insetLog(1, $task, "开始采集热榜数据 ",$task_run_log_id);
            // 检查用户的豆子余额
            $this->checkBeans($task,2,$task_run_log_id);
            // 开始采集任务
            $this->dataMatch($task,$task_run_log_id);
            // 根据采集结果进行处理
            if($task['text_select']==1){
                // 如果选择了文本匹配，则查询匹配结果并更新状态
                $out_redian_data = RedianData::mQuery()->where(['task_id'=>$task_id,"task_run_log_id"=>$task_run_log_id])
                ->where('similarity','>',79)->order('similarity desc')->limit(0,$task['out_num'])->select()->toArray();
                $out_redian_data_ids = array_column($out_redian_data,'id');
                $r = RedianData::mQuery()->whereIn('id',$out_redian_data_ids)->update(['filter_status'=>1]);
                if($r){
                    $progress=3;
                }else{
                    // 如果采集失败，则更新任务错误信息
                    RedianTask::mQuery()->where([ "id"=>$task_id])->update(['error_msg'=>"采集失败:无匹配采集结果",'error_status'=>1,'status'=>1]);
                    if($task_run_log_id){
                        $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['result'=>"采集失败:无匹配采集结果",'error_status'=>1]);
                    }
                    $this->setQueueError("任务执行失败:无匹配采集结果");
                }
            }else{
                // 如果未选择文本匹配，则直接检查采集结果数量
                $out_redian_data_num = RedianData::mQuery()->where(['task_id'=>$task_id,"task_run_log_id"=>$task_run_log_id])->count();
                if($out_redian_data_num>0){
                    // 如果有采集结果，则更新任务进度
                    RedianTask::mQuery()->where([ "id"=>$task_id])->update(['progress'=>2]);
                    if($task_run_log_id){
                        $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['progress'=>2]);
                    }
                    $this->setQueueSuccess("采集完成,等待用户筛选文案");
                }else{
                    // 如果没有采集结果，则更新任务错误信息
                    RedianTask::mQuery()->where([ "id"=>$task_id])->update(['error_msg'=>"采集失败:无匹配采集结果",'error_status'=>1,'status'=>1]);
                    if($task_run_log_id){
                        $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['result'=>"采集失败:无匹配采集结果",'error_status'=>1]);
                    }
                    $this->setQueueError("任务执行失败:无采集结果");
                }
            }
        }

        // 如果任务进度为3，则执行文本生成步骤
        if( $progress==3){
            // 检查用户的豆子余额
            $this->checkBeans($task,3,$task_run_log_id);
            // 口播文本生成
            $this->setQueueProgress("开始口播文本生成,请稍等 ", 40);
            RedianTask::mQuery()->where([ "id"=>$task_id])->update(['progress'=>3]);
            if($task_run_log_id){
                $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['progress'=>3]);
            }
            // 循环执行文本生成，直到成功或超过最大尝试次数
            $f = true;
            $f_num = 0;
            while ($f){
                $f = $this->textGenerate($task,$task_run_log_id);
                $f_num++;
                if($f_num>5){
                    $f = false;
                }
            }
            // 检查生成结果
            $c = RedianData::mQuery()->where(['task_id'=>$task['id'],'gaixie_status'=>2,'task_run_log_id'=>$task_run_log_id])->count();
            if($c==0){
                // 如果生成失败，则更新任务错误信息
                RedianTask::mQuery()->where([ "id"=>$task_id])->update(['error_msg'=>"AI改写失败",'error_status'=>3,'status'=>1]);
                if($task_run_log_id){
                    $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['result'=>"AI改写失败",'error_status'=>3]);
                }
                $this->setQueueError("任务执行失败:AI改写失败");
            }
            // 更新任务进度
            $progress=4;
        }

        // 如果任务进度为4，则执行音频生成步骤
        if($progress==4){
            // 检查用户的豆子余额
            $this->checkBeans($task,4,$task_run_log_id);
            // 音频生成
            $this->setQueueProgress("开始音频生成,请稍等 ", 60);
            RedianTask::mQuery()->where([ "id"=>$task_id])->update(['progress'=>4]);
            if($task_run_log_id){
                $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['progress'=>4]);
            }
            // 执行音频生成并检查结果
            $redian_data = TaskService::instance()->audioGenerate($task,$task_run_log_id);
            if(empty($redian_data)){
                // 如果生成失败，则更新任务错误信息
                RedianTask::mQuery()->where([ "id"=>$task_id])->update(['error_msg'=>"音频生成失败",'error_status'=>4,'status'=>1]);
                if($task_run_log_id){
                    $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['result'=>"音频生成失败",'error_status'=>4]);
                }
                $this->setQueueError("任务执行失败:音频生成失败111");
            }
            // 如果使用蝉镜生成音频，则需要等待音频生成完成
            if($task['audio_platform']=="蝉镜"){
                $audio_result = false;
                while (!$audio_result){
                    $this->setQueueProgress("等待音频生成结果{$audio_result}");
                    $this->insetLog(1,$task_id,"WaveNet 音频合成中 ",$task_run_log_id);
                    sleep(60);
                    // 还有正在生成中的音频生成任务继续等待
                    $mun = RedianData::mQuery()->where(["task_id"=>$task_id,"task_run_log_id"=>$task_run_log_id])->where(['audio_status'=>2])->count();
                    if($mun>0){
                        $this->setQueueProgress("有音频生成任务正在处理中 {$mun}");
                    }else{
                        $audio_result = true;
                        $this->setQueueProgress("音频生成完成");
                    }
                }
                sleep(30);
                $r = RedianData::mQuery()->where(["task_id"=>$task_id,"task_run_log_id"=>$task_run_log_id])->where(['audio_status'=>3])->count();
                $this->setQueueProgress("音频生成结果{$r}");
                if($r==0){
                    // 如果生成失败，则更新任务错误信息
                    RedianTask::mQuery()->where([ "id"=>$task_id])->update(['error_msg'=>"音频生成失败",'error_status'=>4,'status'=>1]);
                    if($task_run_log_id){
                        $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['result'=>"音频生成失败",'error_status'=>4]);
                    }
                    $this->setQueueError("任务执行失败:音频生成失败{$r}");
                }
            }
            // 更新任务进度
            $progress=5;
        }

        // 如果任务进度为5，则执行视频生成步骤
        if($progress == 5){
            // 检查用户的豆子余额
            $this->checkBeans($task,5,$task_run_log_id);
            // 视频生成
            $this->setQueueProgress("开始视频生成,请稍等 ", 80);
            RedianTask::mQuery()->where([ "id"=>$task_id])->update(['progress'=>5]);
            if($task_run_log_id){
                $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['progress'=>5]);
            }
            // 执行视频生成并检查结果
            $redian_data = TaskService::instance()->videoGenerate($task,$task_run_log_id);
            if(empty($redian_data)){
                // 如果生成失败，则更新任务错误信息
                RedianTask::mQuery()->where([ "id"=>$task_id])->update(['error_msg'=>"视频生成失败",'error_status'=>5,'status'=>1]);
                if($task_run_log_id){
                    $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['result'=>"视频生成失败",'error_status'=>5]);
                }
                $this->setQueueError("任务执行失败:视频生成失败");
            }
            // 如果使用HeyGen平台生成视频，则创建视频生成状态查询任务
            if($task['video_platform']=="HeyGen"){
                // 创建视频生成状态查询任务
                sysqueue($task['title']."视频生成状态查询任务", "xbase:heygenRetrieve",  10, $data = [
                    'task_id'=>$task_id,'task_run_log_id'=>$task_run_log_id
                ], $rscript = 1, $loops = 0);
            }
            if($task['video_platform']=="小甲鱼"){
                // 创建视频生成状态查询任务
                sysqueue($task['title']."视频生成状态查询任务(小甲鱼)", "xbase:XiaojiayuRetrieve",  10, $data = [
                    'task_id'=>$task_id,'task_run_log_id'=>$task_run_log_id
                ], $rscript = 1, $loops = 0);
            }
        }

        // 任务完成
        $this->setQueueProgress("任务完成", 100);
        $this->setQueueSuccess("任务完成");
    }


    //AI改写处理方法
    public function textGenerate($task,$task_run_log_id)
    {
        //查询已改写数量
        $c = RedianData::mQuery()->where(['task_id'=>$task['id'],'gaixie_status'=>2,'task_run_log_id'=>$task_run_log_id])->count();
        if($c>=$task['out_num']){
            return false;
        }
        if($c==0){
            $redian_data = TaskService::instance()->textGenerate($task,$task_run_log_id);
            if($redian_data<$task['out_num']){
                return true;
            }else{
                return false;
            }
        }else{
            $out_redian_data = RedianData::mQuery()->where(['task_id'=>$task['id'],"task_run_log_id"=>$task_run_log_id,'filter_status'=>0])
                ->where('similarity','>',0)->order('similarity desc')->limit(0,($task['out_num']-$c))->select()->toArray();
            if(!empty($out_redian_data)){
                $out_redian_data_ids = array_column($out_redian_data,'id');
                RedianData::mQuery()->whereIn('id',$out_redian_data_ids)->update(['filter_status'=>1]);
                $redian_data = TaskService::instance()->textGenerate($task,$task_run_log_id);
                if($redian_data<$task['out_num']){
                    return true;
                }else{
                    return false;
                }
            }else{
                return false;
            }
        }


    }


    /**
     * 执行任务数据匹配与采集流程
     *
     * @param array $task 任务配置数组，需包含任务ID、输出数量、节点配置等信息
     * @param int|null $task_run_log_id 任务运行日志ID，可为空
     * @return int 固定返回1，表示流程执行完成(实际可能需要根据业务需求调整返回值含义)
     */
    public function dataMatch($task,$task_run_log_id)
    {
        // 更新任务进度为"进行中"状态
        $task_id = $task['id'];
        RedianTask::mQuery()->where([ "id"=>$task_id])->update(['progress'=>1]);
        if($task_run_log_id){
            $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['progress'=>1]);
        }

        // 初始化采集控制参数
        $max_data_num = $task['out_num']*10; // 最大采集量=输出需求*10倍
        $data_num = 0;    // 已采集计数器
        $execute_num =1;  // 执行次数计数器
        $flag = true;     // 流程控制标志

        // 获取热点节点列表(排除任务中已经采集过的节点，空值取全量节点)
        $task_nodes = $task['task_nodes'];
        $hot_nodes = $this->app->db->name("redian_hot_nodes")->whereIn("node",$task_nodes)->order("sort desc")->select()->toArray();
        if(empty($hot_nodes)){
            $hot_nodes = $this->app->db->name("redian_hot_nodes")->order("sort desc")->select()->toArray();
        }

        // 节点处理主循环
        $zc_nodes = ["KqndgxeLl9","L4MdA5ldxD","K7GdaMgdQy"]; // 特殊节点白名单(无法通过coze采集的节点)
        $yc_nodes = []; // 已采集节点记录
        foreach ($hot_nodes as $hot_node){
            $this->setQueueProgress("开始采集节点{$hot_node['title']}请稍等 ", 0);
            $this->insetLog(1,$task_id,"开始采集{$hot_node['title']}热榜数据,请稍等 ");
            // 节点分类处理逻辑
            if(in_array($hot_node['node'],$zc_nodes)){
                // 白名单节点直接采集
                $r = TaskService::instance()->startGather($task,$task_run_log_id,$hot_node['node']);
                if($r==-1 && $execute_num==1){
                    RedianTask::mQuery()->where([ "id"=>$task_id])->update(['error_msg'=>"采集失败:蓝豆不足",'error_status'=>1,'status'=>1]);
                    if($task_run_log_id){
                        $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['result'=>"采集失败:蓝豆不足",'error_status'=>1]);
                    }
                }
            }else{
                // 普通节点启动采集工作流
                $resCozeNode = TaskService::instance()->startGatherCozeNode($task,1,$hot_node['node']);
                sleep(10); // 等待工作流初始化

                // 工作流结果轮询机制
                $execute_id = $resCozeNode['data'];
                $i = 0;
                $WorkflowResult  = 0;
                while ( $i<$max_data_num){
                    $this->insetLog(1,$task_id,"查询采集结果,请稍等 ",$task_run_log_id);
                    $this->setQueueProgress("查询采集结果,请稍等 ".$execute_id, 0);
                    $WorkflowResult = $res = TaskService::instance()->getWorkflowResult($task,$execute_id,$task_run_log_id,7511747390979571749);

                    // 处理不同采集状态
                    if($res==0){
                       break;
                    }elseif ($res==1){
                        //扣除蓝豆
                        $ldr = MemberService::instance()->charge($task['user_id'],"ByRdGather",$task);
                        if($ldr['code']==0){
                            RedianTask::mQuery()->where([ "id"=>$task_id])->update(['error_msg'=>"采集失败:扣除蓝豆失败".$ldr['msg'],'error_status'=>1,'status'=>1]);
                            if($task_run_log_id){
                                $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['result'=>"采集失败:扣除蓝豆失败".$ldr['msg'],'error_status'=>1]);
                            }
                            $this->setQueueError("扣除蓝豆失败".$ldr['msg'], 0);
                        }
                        break; // 成功状态跳出循环
                    }elseif($res==3){
                        $flag = false; // 无数据标志
                        break;
                    }else{
                        sleep(60); // 轮询间隔
                    }
                    $i++;
                }
            }

            // 后处理流程
            $yc_nodes[]=$hot_node['id'];
            if($task['text_select']==1){
                $this->similarity($task,$task_run_log_id); // 相似度筛选
            }

            // 结果集检查与退出判断
            $redian_query = RedianData::mQuery()->where(['task_id'=>$task_id,"task_run_log_id"=>$task_run_log_id]);
            $yes_num =$max_data_num;
            if($task['text_select']==1){
                $redian_query->where('similarity','>',79); // 相似度过滤
                $yes_num = $task['out_num'];
            }
            $out_redian_data = $redian_query->order('similarity desc')->select()->toArray();

            if($yes_num<=count($out_redian_data)){
                break; // 达到采集量要求时终止
            }
            $execute_num++;
        }

        // 最终状态更新
        RedianTask::mQuery()->where([ "id"=>$task['id']])->update(['hot_nodes'=>implode(",",$yc_nodes)]);
        return 1;
    }


    public function checkBeans($task,$progress,$task_run_log_id)
    {
        $beans = MemberService::instance()->getUserRemainingBeans($task['user_id']);
        if($beans<2){
            RedianTask::mQuery()->where([ "id"=>$task['id']])->update(['error_msg'=>"蓝豆不足",'error_status'=>$progress,'status'=>1]);
            if($task_run_log_id){
                $this->app->db->name("task_run_log")->where(["id"=>$task_run_log_id])->update(['result'=>"蓝豆不足"]);
            }
            $this->setQueueError("任务执行失败:蓝豆不足");
        }
    }


    /**
     * 计算任务数据相似度并处理结果
     *
     * @param array $task 任务信息数组，需包含'id'字段表示任务ID
     * @param int $task_run_log_id 任务运行日志记录ID
     * @return int 返回处理状态：1-成功，0-失败
     */
    public function similarity($task,$task_run_log_id)
    {
        $task_id = $task['id'];
        // 查询未计算相似度的数据量
        $r = RedianData::mQuery()->where(['task_id'=>$task['id'],"task_run_log_id"=>$task_run_log_id])->whereNull('similarity')->count();

        if($r>0){
            $this->insetLog(1,$task_id,"COSINE 文案匹配度计算中",$task_run_log_id);
            // 存在未处理数据时进入处理流程
            $this->setQueueProgress("计算匹配度,请稍等 ", 0);
            RedianTask::mQuery()->where([ "id"=>$task_id])->update(['progress'=>2]);

            // 批量相似度计算尝试（最多5次）
            $batch_similarity_num = 1;
            $batch_id = 0;
            while ($batch_similarity_num<5){
                $b_r = TaskService::instance()->batchSimilarity($task,$task_run_log_id);
                if($b_r){
                    $batch_id = $b_r;
                    break;
                }
                $batch_similarity_num++;
                sleep(30);
            }

            if($batch_id==0){
                return 0;
            }

            // 相似度结果查询流程（最多15次）
            $batch_similarity_result_num = 1;
            $similarity_result = 0;
            $this->setQueueProgress("查询计算匹配度,请稍等 ", 0);
            while ($batch_similarity_result_num<15 && $batch_id!=0){
                $this->setQueueProgress("第{$batch_similarity_result_num}次查询计算匹配度,请稍等 ", 0);
                $this->insetLog(1,$task_id,"COSINE 文案匹配度计算中",$task_run_log_id);
                sleep(60);
                $similarity_result = TaskService::instance()->batchSimilarityResult($task['user_id'],$batch_id);
                $n= json_encode($similarity_result);
                $this->setQueueProgress("第{$batch_similarity_result_num}次查询计算匹配度:{$n}", 0);

                // 根据返回结果决定是否终止查询
                if(($similarity_result==0 || $similarity_result==3)){
                    break;
                }

                if($similarity_result==2){
                    break;
                }
                $batch_similarity_result_num++;
            }

            // 处理批量任务超时的情况
            if($similarity_result==1){
                $this->setQueueProgress("批量任务一直在等待中,则取消该批量任务,改为单个任务,请稍等 ", 0);
                TaskService::instance()->cancelBatch($task['user_id'],$batch_id);
                $redian_data = TaskService::instance()->similarity($task,$task_run_log_id);
                if(empty($redian_data)){
                    return 0;
                }
            }
        }
        return 1;
    }
    public function insetLog($type,$task_id,$msg='',$task_run_log_id=0)
    {
        $this->app->db->name("task_log")->insert([
            'type'=>$type,
            'task_id'=>$task_id,
            'task_run_log_id'=>$task_run_log_id,
            'content'=>$msg
        ]);
    }



}
